package jar

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

import scala.collection.mutable.Map

object SparkOperator {

  var sparkConfig=new SparkConf().setMaster("local").setAppName("sparkRddTest")

  val sparkContext =new SparkContext(sparkConfig)

  def main(args: Array[String]): Unit = {
    //日志显示级别,设置打印error级别的错误日志
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.ERROR)

//    testBaseTransformation

    testMapValue

  }

  def testBaseTransformation(): Unit ={
    // 创建rdd 数据范围 1-5整数，设置15个partitions
    var rdd = sparkContext.makeRDD(1 to 50, 15)

    // 设置partition的优先位置
    rdd.preferredLocations(rdd.partitions(0))
    rdd.preferredLocations(rdd.partitions(1))

    // 使用collect 合并partitions ,之后设置权重调用randomSplit函数将rdd按权重随机分配
    rdd.collect
    var splitRDD = rdd.randomSplit(Array(0.05, 0.25, 0.15, 0.35, 0.1, 0.1))
    var splitRDDCollect=splitRDD(0).collect
    splitRDDCollect=splitRDD(1).collect
    splitRDDCollect=splitRDD(2).collect
    splitRDDCollect=splitRDD(3).collect
    splitRDDCollect=splitRDD(4).collect


    println(rdd.partitions.size)

    // 统计每个 partition 中对应的元素
    var rddcollect=
      rdd.mapPartitionsWithIndex {
        (index, iter) => {
          val resultMap = Map[String, List[Int]]()
          while (iter.hasNext) {
            var part_name = "part_" + index
            var elem = iter.next()
            if (resultMap.contains(part_name)) {
              var elems = resultMap(part_name)
              elems ::= elem
              resultMap(part_name) = elems
            } else {
              resultMap(part_name) = List[Int] {
                elem
              }
            }
          }
          resultMap.iterator
        }
      }.collect

    //统计每个分区中的元素个数
    var rddcollect2=
      rdd.mapPartitionsWithIndex(
        (index,iter)=>{
          var partMap=Map[String,Int]()
          var partName="part_"+index
          while (iter.hasNext){
            if(partMap.contains(partName)){
              partMap(partName)=partMap(partName)+1
            }else{
              partMap(partName)=1
            }
          }
          partMap.iterator

        }
      )


    println(rddcollect)

  }

  def testMapValue(): Unit ={

    var rdd = sparkContext.makeRDD(Array((1,'A'),(2,'B'),(3,'C'),(4,'D'),(5,'E'),(6,'F')),2)



    rdd.mapValues(_+"$").collect.foreach(x=>println(x._1+" "+x._2))

    val x=getPartitionElemsFromRDD(rdd.partitionBy(new HashPartitioner(2))).collect

    println(x)

  }

  // scala不推荐使用返回值，因为这样是函数式编程范式被返回值类型约束起来，
  def getPartitionElemsFromRDD[T](rdd: RDD[T]) = {
    rdd.mapPartitionsWithIndex {
      (index, iter) => {
        val resultMap = Map[String, List[T]]()
        while (iter.hasNext) {
          var part_name = "part_" + index
          var elem = iter.next()
          if (resultMap.contains(part_name)) {
            var elems = resultMap(part_name)
            elems ::= elem
            resultMap(part_name) = elems
          } else {
            resultMap(part_name) = List[T] (elem)

          }
        }
        resultMap.iterator
      }
    }
  }

}
